package com.ruoyi.project.app.queue;

import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonDelayedQueue;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Slf4j
//@Service
public class RedisDelayedQueue {

    @Autowired
    private RedissonClient redissonClient;

    /**
     *
     * @param t 泛型的参数
     * @param delayMillis 延迟多少ms
     * @param <T>
     */
    public <T> void produce(T t, String businessName,long delayMillis) {
        produce(t,businessName,delayMillis,TimeUnit.MILLISECONDS);
    }

    /**
     * 添加队列
     *
     * @param t        DTO传输类
     * @param delay    时间数量
     * @param timeUnit 时间单位
     * @param <T>      泛型
     */
    public <T> void produce(T t, String businessName,long delay, TimeUnit timeUnit) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(businessName);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
        delayedQueue.destroy();
    }

    public <T> boolean remove(T t,String businessName){
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(businessName);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        boolean removeResult = delayedQueue.remove(t);
        delayedQueue.destroy();
        return removeResult;
    }



    /**
     * 获取队列
     *
     * @param businessName
     * @param callback 任务回调监听
     * @param <T>   泛型
     * @return
     */
    public <T> void consume(String businessName, RedisDelayedQueueCallback callback) {

        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(businessName);

        new Thread(()->{
            while (true) {
                try {
                    T t = blockingFairQueue.take();
                    callback.callback(t);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }
}
